package com.atguigu.chapter06;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashSet;
import java.util.Set;

/**
 * Author: Pepsi
 * Date: 2023/7/31
 * Desc:
 */
public class Flink03_Project_UV {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port",1000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);

        env
                .readTextFile("input/UserBehavior.csv")
                // 过滤出 PV 数据
                .filter(new FilterFunction<String>() {
                    @Override
                    public boolean filter(String value) throws Exception {
                        return "pv".equals(value.split(",")[3]);
                    }
                })
                // 将PV的数据分到同一个组里面
                .keyBy(new KeySelector<String, String>() {
                    @Override
                    public String getKey(String value) throws Exception {
                        return value.split(",")[3];
                    }
                })
                // 对同一组里面的数据进行处理
                .process(new ProcessFunction<String, String>() {
                    // 存放 userId 的容器，回自动对数据进行去重，最后直接拿它的大小就知道UV了
                    Set<Long> userIdSet = new HashSet<>();
                    @Override
                    public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
                        Long userId = Long.valueOf(value.split(",")[0]);
                        // 向set中添加userId，判断是否添加成功
                        if (userIdSet.add(userId)) {
                            // 添加成功的话，说明是一个新的独立访客，收集到此时容器大小
                            out.collect("UV = "+ userIdSet.size());
                        }
                    }
                })
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
